use criterion::{criterion_group, criterion_main, Criterion, BenchmarkId};
use rustcloud_core::{Request, ServiceInstance, ServiceRegistry, LoadBalancerStrategy};
use rustcloud_registry::InMemoryRegistry;
use rustcloud_loadbalancer::LoadBalancerManager;
use rustcloud_transport::HttpTransport;
use rustcloud_proxy::ServiceProxy;
use rustcloud_resilience::{RetryExecutor, RetryConfig, BackoffStrategy};
use rustcloud_observability::{MetricsCollector, HealthCheckManager};
use rustcloud_bus::memory::InMemoryMessageBus;
use rustcloud_stream::{BusStreamSource, BusStreamSink, MapProcessor, FilterProcessor};

use std::sync::Arc;
use std::time::Duration;
use tokio::runtime::Runtime;
use serde_json::json;

/// 服务注册和发现性能基准测试
fn benchmark_service_registry(c: &mut Criterion) {
    let rt = Runtime::new().unwrap();
    
    c.bench_function("service_registry_register", |b| {
        b.to_async(&rt).iter(|| async {
            let registry = InMemoryRegistry::new();
            let instance = ServiceInstance::new("test-service", "localhost", 8080, "http");
            registry.register(instance).await.unwrap();
        });
    });
    
    c.bench_function("service_registry_discover", |b| {
        b.to_async(&rt).iter_setup(
            || {
                let rt = Runtime::new().unwrap();
                rt.block_on(async {
                    let registry = InMemoryRegistry::new();
                    let instance = ServiceInstance::new("benchmark-service", "localhost", 8080, "http");
                    registry.register(instance).await.unwrap();
                    registry
                })
            },
            |registry| async move {
                registry.discover("benchmark-service").await.unwrap();
            }
        );
    });
    
    // 批量操作性能测试
    let mut group = c.benchmark_group("service_registry_batch");
    for size in [10, 50, 100, 500].iter() {
        group.bench_with_input(BenchmarkId::new("batch_register", size), size, |b, &size| {
            b.to_async(&rt).iter(|| async move {
                let registry = InMemoryRegistry::new();
                for i in 0..size {
                    let instance = ServiceInstance::new(
                        &format!("service-{}", i),
                        "localhost",
                        8080 + i as u16,
                        "http"
                    );
                    registry.register(instance).await.unwrap();
                }
            });
        });
    }
    group.finish();
}

/// 负载均衡性能基准测试
fn benchmark_load_balancer(c: &mut Criterion) {
    let rt = Runtime::new().unwrap();
    
    let mut group = c.benchmark_group("load_balancer");
    
    // 测试不同实例数量下的负载均衡性能
    for instance_count in [5, 10, 50, 100].iter() {
        let instances: Vec<ServiceInstance> = (0..*instance_count)
            .map(|i| ServiceInstance::new("service", &format!("host-{}", i), 8080, "http"))
            .collect();
        
        group.bench_with_input(
            BenchmarkId::new("round_robin", instance_count),
            &instances,
            |b, instances| {
                let lb = LoadBalancerManager::new(LoadBalancerStrategy::RoundRobin);
                b.iter(|| {
                    lb.select(instances).unwrap();
                });
            }
        );
        
        group.bench_with_input(
            BenchmarkId::new("random", instance_count),
            &instances,
            |b, instances| {
                let lb = LoadBalancerManager::new(LoadBalancerStrategy::Random);
                b.iter(|| {
                    lb.select(instances).unwrap();
                });
            }
        );
    }
    group.finish();
}

/// 容错机制性能基准测试
fn benchmark_resilience(c: &mut Criterion) {
    let rt = Runtime::new().unwrap();
    
    c.bench_function("retry_executor_success", |b| {
        b.to_async(&rt).iter(|| async {
            let config = RetryConfig {
                max_attempts: 3,
                timeout: Duration::from_secs(1),
                backoff: BackoffStrategy::Fixed { delay: Duration::from_millis(10) },
                retry_conditions: vec![],
                budget_percent: 10.0,
                jitter_enabled: false,
            };
            let executor = RetryExecutor::new(config);
            
            executor.execute(|| async { Ok::<String, rustcloud_resilience::ResilienceError>("success".to_string()) }).await.unwrap();
        });
    });
    
    c.bench_function("retry_executor_with_retries", |b| {
        b.to_async(&rt).iter(|| async {
            let config = RetryConfig {
                max_attempts: 3,
                timeout: Duration::from_secs(1),
                backoff: BackoffStrategy::Fixed { delay: Duration::from_millis(1) },
                retry_conditions: vec![],
                budget_percent: 10.0,
                jitter_enabled: false,
            };
            let executor = RetryExecutor::new(config);
            
            let mut attempt = 0;
            let _result = executor.execute(|| async {
                attempt += 1;
                if attempt < 3 {
                    Err(rustcloud_resilience::ResilienceError::operation_failed("fail".to_string()))
                } else {
                    Ok("success".to_string())
                }
            }).await;
        });
    });
}

/// 可观测性性能基准测试
fn benchmark_observability(c: &mut Criterion) {
    let rt = Runtime::new().unwrap();
    
    c.bench_function("metrics_collector_counter", |b| {
        b.to_async(&rt).iter(|| async {
            let collector = MetricsCollector::new();
            collector.record_counter("test.counter", 1.0, vec![]).await;
        });
    });
    
    c.bench_function("metrics_collector_histogram", |b| {
        b.to_async(&rt).iter(|| async {
            let collector = MetricsCollector::new();
            collector.record_histogram("test.histogram", 100.0, vec![]).await;
        });
    });
    
    c.bench_function("health_check_manager", |b| {
        b.to_async(&rt).iter(|| async {
            let manager = HealthCheckManager::new();
            manager.check_health().await.unwrap();
        });
    });
    
    // 批量指标记录性能测试
    let mut group = c.benchmark_group("metrics_batch");
    for batch_size in [100, 500, 1000].iter() {
        group.bench_with_input(BenchmarkId::new("batch_counters", batch_size), batch_size, |b, &size| {
            b.to_async(&rt).iter(|| async move {
                let collector = MetricsCollector::new();
                for i in 0..size {
                    collector.record_counter(&format!("test.counter.{}", i), 1.0, vec![]).await;
                }
            });
        });
    }
    group.finish();
}

/// 消息总线性能基准测试
fn benchmark_message_bus(c: &mut Criterion) {
    let rt = Runtime::new().unwrap();
    
    c.bench_function("message_bus_publish", |b| {
        b.to_async(&rt).iter_setup(
            || {
                let rt = Runtime::new().unwrap();
                rt.block_on(async {
                    InMemoryMessageBus::new().await.unwrap()
                })
            },
            |bus| async move {
                let publisher = bus.create_publisher().unwrap();
                let message = rustcloud_bus::prelude::MessageBuilder::new()
                    .topic("test.topic")
                    .payload(json!({"test": "data"}))
                    .build();
                publisher.publish(message).await.unwrap();
            }
        );
    });
    
    c.bench_function("message_bus_subscribe", |b| {
        b.to_async(&rt).iter_setup(
            || {
                let rt = Runtime::new().unwrap();
                rt.block_on(async {
                    InMemoryMessageBus::new().await.unwrap()
                })
            },
            |bus| async move {
                let _subscriber = bus.create_subscriber("test-sub", "test.*").await.unwrap();
            }
        );
    });
    
    // 高吞吐量测试
    let mut group = c.benchmark_group("message_bus_throughput");
    for message_count in [100, 500, 1000].iter() {
        group.bench_with_input(BenchmarkId::new("publish_batch", message_count), message_count, |b, &count| {
            b.to_async(&rt).iter_setup(
                || {
                    let rt = Runtime::new().unwrap();
                    rt.block_on(async {
                        InMemoryMessageBus::new().await.unwrap()
                    })
                },
                |bus| async move {
                    let publisher = bus.create_publisher().unwrap();
                    for i in 0..count {
                        let message = rustcloud_bus::prelude::MessageBuilder::new()
                            .topic(&format!("test.topic.{}", i))
                            .payload(json!({"id": i, "data": "test"}))
                            .build();
                        publisher.publish(message).await.unwrap();
                    }
                }
            );
        });
    }
    group.finish();
}

/// 流处理性能基准测试
fn benchmark_stream_processing(c: &mut Criterion) {
    let rt = Runtime::new().unwrap();
    
    c.bench_function("stream_processor_map", |b| {
        b.to_async(&rt).iter(|| async {
            let processor = MapProcessor::with_mapper(|mut msg| {
                let mut payload = msg.payload().clone();
                payload["processed"] = json!(true);
                msg.set_payload(payload);
                msg
            });
            
            let message = rustcloud_bus::prelude::MessageBuilder::new()
                .topic("test")
                .payload(json!({"original": "data"}))
                .build();
            
            processor.process(message).await.unwrap();
        });
    });
    
    c.bench_function("stream_processor_filter", |b| {
        b.to_async(&rt).iter(|| async {
            let processor = FilterProcessor::with_predicate(|msg| {
                msg.payload().get("active").and_then(|v| v.as_bool()).unwrap_or(true)
            });
            
            let message = rustcloud_bus::prelude::MessageBuilder::new()
                .topic("test")
                .payload(json!({"active": true, "data": "test"}))
                .build();
            
            processor.process(message).await.unwrap();
        });
    });
    
    // 流处理管道性能测试
    let mut group = c.benchmark_group("stream_pipeline");
    for message_count in [50, 100, 200].iter() {
        group.bench_with_input(BenchmarkId::new("full_pipeline", message_count), message_count, |b, &count| {
            b.to_async(&rt).iter(|| async move {
                let bus = Arc::new(InMemoryMessageBus::new().await.unwrap());
                let mut source = BusStreamSource::with_defaults(bus.clone(), "input.*").await.unwrap();
                let mut sink = BusStreamSink::with_defaults(bus.clone(), "output.processed").await.unwrap();
                
                let map_processor = MapProcessor::with_mapper(|mut msg| {
                    let mut payload = msg.payload().clone();
                    payload["processed_at"] = json!(chrono::Utc::now().timestamp());
                    msg.set_payload(payload);
                    msg
                });
                
                let filter_processor = FilterProcessor::with_predicate(|msg| {
                    msg.payload().get("valid").and_then(|v| v.as_bool()).unwrap_or(true)
                });
                
                source.start().await.unwrap();
                
                // 发布测试消息
                let publisher = bus.create_publisher().unwrap();
                for i in 0..count {
                    let message = rustcloud_bus::prelude::MessageBuilder::new()
                        .topic(&format!("input.event.{}", i))
                        .payload(json!({"id": i, "valid": true, "data": "test"}))
                        .build();
                    publisher.publish(message).await.unwrap();
                }
                
                // 处理消息
                let mut processed = 0;
                while processed < count {
                    if let Some(Ok(message)) = source.next().await {
                        if filter_processor.should_process(&message) {
                            let mapped = map_processor.process(message).await.unwrap();
                            sink.send(mapped).await.unwrap();
                            processed += 1;
                        }
                    }
                }
                
                sink.flush().await.unwrap();
                source.stop().await.unwrap();
                sink.close().await.unwrap();
            });
        });
    }
    group.finish();
}

/// 端到端性能基准测试
fn benchmark_end_to_end(c: &mut Criterion) {
    let rt = Runtime::new().unwrap();
    
    c.bench_function("end_to_end_service_call", |b| {
        b.to_async(&rt).iter_setup(
            || {
                let rt = Runtime::new().unwrap();
                rt.block_on(async {
                    let registry = Arc::new(InMemoryRegistry::new());
                    let load_balancer = Arc::new(LoadBalancerManager::new(LoadBalancerStrategy::RoundRobin));
                    let transport = Arc::new(HttpTransport::new());
                    let circuit_breaker = Arc::new(rustcloud_circuit_breaker::DefaultCircuitBreaker::new(
                        rustcloud_circuit_breaker::CircuitBreakerConfig::default()
                    ));
                    
                    let instance = ServiceInstance::new("test-service", "localhost", 8080, "http");
                    registry.register(instance).await.unwrap();
                    
                    ServiceProxy::new(registry, load_balancer, transport, circuit_breaker)
                })
            },
            |proxy| async move {
                let mut request = Request::new("test-service", "GET", "/api/test");
                request.set_body(b"test".to_vec());
                
                let config = rustcloud_core::TransportConfig::default();
                // 注意：这里会失败因为没有真实服务，但我们测试的是框架开销
                let _result = proxy.call(request, &config).await;
            }
        );
    });
}

/// 内存使用效率基准测试
fn benchmark_memory_efficiency(c: &mut Criterion) {
    let rt = Runtime::new().unwrap();
    
    c.bench_function("memory_message_creation", |b| {
        b.iter(|| {
            let _message = rustcloud_bus::prelude::MessageBuilder::new()
                .topic("test.topic")
                .payload(json!({"data": "test", "timestamp": chrono::Utc::now().timestamp()}))
                .build();
        });
    });
    
    c.bench_function("memory_service_instance_creation", |b| {
        b.iter(|| {
            let _instance = ServiceInstance::new("test-service", "localhost", 8080, "http");
        });
    });
    
    // 大对象创建测试
    let mut group = c.benchmark_group("memory_large_objects");
    for size in [100, 500, 1000].iter() {
        group.bench_with_input(BenchmarkId::new("large_payload", size), size, |b, &size| {
            b.iter(|| {
                let payload = json!({
                    "data": (0..size).map(|i| format!("item-{}", i)).collect::<Vec<_>>(),
                    "metadata": {
                        "size": size,
                        "created_at": chrono::Utc::now().timestamp()
                    }
                });
                let _message = rustcloud_bus::prelude::MessageBuilder::new()
                    .topic("test.large")
                    .payload(payload)
                    .build();
            });
        });
    }
    group.finish();
}

criterion_group!(
    benches,
    benchmark_service_registry,
    benchmark_load_balancer,
    benchmark_resilience,
    benchmark_observability,
    benchmark_message_bus,
    benchmark_stream_processing,
    benchmark_end_to_end,
    benchmark_memory_efficiency
);

criterion_main!(benches);